package com.zhang.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * 数据流：web/app -> Nginx -> SpringBoot -> Kafka(ods) -> FlinkApp -> Kafka(dwd)
 * 程  序：mockLog -> Nginx -> Logger.sh  -> Kafka(ZK)  -> BaseLogApp -> kafka
 *
 * @title: ods用户行为数据处理
 * @author: zhang
 * @date: 2022/1/1 10:12
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //1.1 开启CK并指定状态后端为FS    memory  fs  rocksdb
      /*  env.setStateBackend(new FsStateBackend("hdfs://hadoop302:8020/gmall-flink/ck"));
        env.enableCheckpointing(5000L);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
        System.setProperty("HADOOP_USER_NAME", "zhang");*/

        //TODO 2.消费 ods_base_log 主题数据创建流
        String topic = "ods_base_log";
        String groupId = "base_log_app";
        DataStreamSource<String> kafkaDs = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));

        //TODO 3.将每行数据转换为JSON对象 过滤脏数据
        OutputTag<String> dirty = new OutputTag<String>("Dirty") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDs.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    collector.collect(jsonObject);
                } catch (Exception e) {
                    //发生解析异常侧输出流输出
                    context.output(dirty, value);
                }
            }
        });
        jsonObjDS.getSideOutput(dirty).print("dirty>>>>>>>>>");

        //TODO 4.新老用户校验 状态编程
        SingleOutputStreamOperator<JSONObject> jsonWithNewFlagDS = jsonObjDS.keyBy(json -> json.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {

                    //初始化状态
                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        valueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>("new_mid", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {
                        String isNew = value.getJSONObject("common").getString("is_new");
                        if ("1".equals(isNew)) {
                            String state = valueState.value();
                            if (state != null) {
                                value.getJSONObject("common").put("is_new", "0");
                            } else {
                                valueState.update("1");
                            }
                        }
                        return value;
                    }
                });

        //TODO 5.分流 侧输出流 页面：主流  启动：侧输出流  曝光：侧输出流
        OutputTag<String> startOutput = new OutputTag<String>("start") {
        };
        OutputTag<String> displayOutput = new OutputTag<String>("display") {
        };
        SingleOutputStreamOperator<String> pageDS = jsonWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context context, Collector<String> collector) throws Exception {
                String start = value.getString("start");
                if (start != null && start.length() > 0) {
                    //启动日志侧输出流输出
                    context.output(startOutput, value.toJSONString());
                } else {
                    //页面（曝光也属于页面）主流输出
                    collector.collect(value.toString());
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        //获取页面id
                        String pageId = value.getJSONObject("page").getString("page_id");
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            //添加页面id
                            display.put("page_id", pageId);
                            //将输出写到曝光侧输出流
                            context.output(displayOutput, display.toJSONString());
                        }
                    }
                }
            }
        });

        //TODO 6.提取侧输出流数据
        DataStream<String> startDS = pageDS.getSideOutput(startOutput);
        DataStream<String> displayDS = pageDS.getSideOutput(displayOutput);

        //TODO 7.将三个流进行打印并输出到对应的kafka主题中
        startDS.print("start>>>>>>>>");
        pageDS.print("page>>>>>>>>>>");
        displayDS.print("display>>>>>");

        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"));
        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"));

        //TODO 8.启动任务
        env.execute("BaseLogApp");

    }
}
